#include "DatagramServer.h"
#include "DatagramAcceptChannel.h"
#include "DatagramSocketChannel.h"
#include "EventLoop.h"
#include "Manager.h"
#include "PlatLogger.h"
#include "SocketOpt.h"
#include "SubDatagramServer.h"
#include "UdpSocket.h"
#include "Util.h"
#include "XnccAddress.h"
#include "XnccServer.h"
#include "XnccType.h"

namespace xncc {
namespace foundation {

    DatagramServer::DatagramServer(EventLoop*                   loop,
                                   std::shared_ptr<XnccAddress> listenAddr,
                                   std::string                  nameArg,
                                   const std::set<std::string>& blacklist)
        : XnccServer(blacklist), loop_(loop), ipPort_(listenAddr->toStr()), name_(std::move(nameArg))
    {
        if (loop_ == nullptr) {
            std::ostringstream oss;
            oss << " DatagramServer engine  " << name_ << " must has a valid eventloop";
            throw shared::logic_exception(oss.str());
        }
        serverSocket_ = std::make_shared<UdpSocket>(listenAddr);
    }

    DatagramServer::~DatagramServer() = default;

    void DatagramServer::init()
    {
        loop_->registerUserFunc(shared::InnerMsgId::ON_PUB_IPC_MSG,
                                [this](const char* msg, const uint32_t len, const types::thread_id_t tid,
                                       const types::connection_id_t connId) { handleIoMsg(msg, len, tid, connId); });
        SocketOpt::setReuseAddr(true, serverSocket_->fd());
        SocketOpt::setReusePort(true, serverSocket_->fd());
        serverSocket_->bind();
        serverChannel_ = std::make_shared<DatagramAcceptChannel>(loop_, serverSocket_, blacklist_);
        serverChannel_->init();
        serverChannel_->enableReading();
        serverChannel_->setNewConnCallback(
            [this](int sockfd, types::connection_id_t connId, std::shared_ptr<XnccAddress> peerAddr) {
                newConnection(sockfd, connId, std::move(peerAddr));
            });
    }

    void DatagramServer::start() { TRACE_INFO << "engine id:"_s << loop_->id() << " server started:"_s << ipPort_; }

    void
    DatagramServer::handleIoMsg(const char* msg, uint32_t len, types::thread_id_t tid, types::connection_id_t connId)
    {
        if (len > MAX_UDP_MSG_LEN) {
            throw shared::logic_exception(" msg too bigger");
            return;
        }
        auto ptr = clientChs_[connId];
        if (ptr == nullptr) {
            TRACE_ERROR << "msg sender set error connid:"_s << connId;
            return;
        }
        ptr->sendMsg(msg, len);
    }

    void DatagramServer::createSubServer()
    {
        for (auto* loop : subServerLoops_) {
            auto subserver = std::make_shared<SubDatagramServer>(loop);
            subServers_.push_back(subserver);
        }
    }

    void DatagramServer::newConnection(int sockfd, types::connection_id_t connId, std::shared_ptr<XnccAddress> peerAddr)
    {
        if (subServers_.size() > 0) {
            if (nextSubserverIndex_ == subServers_.size()) {
                handleNewConnection(sockfd, connId, peerAddr);
                nextSubserverIndex_ = 0;
            }
            else {
                subServers_[nextSubserverIndex_]->acceptNewConnection(sockfd, connId, peerAddr);
                ++nextSubserverIndex_;
            }
        }
        else {
            handleNewConnection(sockfd, connId, peerAddr);
        }
    }

    void DatagramServer::handleNewConnection(int                          sockfd,
                                             types::connection_id_t       connId,
                                             std::shared_ptr<XnccAddress> peerAddr)
    {
        // 构建消息
        const std::string peerAddrStr = peerAddr->toStr();
        TRACE_INFO << " engine id:"_s << loop_->id() << " new connection, connid:"_s << connId << " peeraddr:"_s
                   << peerAddrStr;
        QUEUE_MGR.pubIPCDataToBusiThread(peerAddrStr.c_str(), static_cast<uint32_t>(peerAddrStr.size()),
                                         shared::InnerMsgId::ON_CONNECTED, loop_->id(), connId);

        // 一个连接建立之后，投递连接信息到 关注消息的线程， 然后要把这个fd，放进epool循环中
        auto chPtr         = std::make_shared<DatagramSocketChannel>(loop_, sockfd, connId);
        clientChs_[connId] = chPtr;
        chPtr->init();
        chPtr->setCloseCallback([this, connId](const std::string& reason) { removeConnection(connId, reason); });
        chPtr->enableReading();
    }

    void DatagramServer::removeConnection(types::connection_id_t connId, const std::string& disconnectedReason)
    {
        auto iter = clientChs_.find(connId);
        if (iter != clientChs_.end()) {
            TRACE_INFO << " engine id:"_s << loop_->id() << " connection disconnected, connid:"_s << connId;
            QUEUE_MGR.pubIPCDataToBusiThread(disconnectedReason.c_str(),
                                             static_cast<uint32_t>(disconnectedReason.size()),
                                             shared::InnerMsgId::ON_DISCONNECTED, loop_->id(), connId);
            iter->second->disableAll();
            iter->second->remove();
            clientChs_.erase(connId);
        }
        else {
            TRACE_INFO << "client disconnect but not founnd confd, fd = "_s << connId;
        }
    }

}  // namespace foundation
}  // namespace xncc